package com.atguigu.app.dwd.db;

import com.atguigu.util.KafkaUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class DwdTradeCartAdd {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tableEnvironment = StreamTableEnvironment.create(env);
        String topicName="topic_db";
        String groupId="dwd_trade_cart_add";
        tableEnvironment.executeSql(KafkaUtil.getCreateTopicDB()+ KafkaUtil.getKafkaDDL(topicName,groupId));
        //tableEnvironment.sqlQuery("select * from topic_db").execute().print();
        Table cartAddTable = tableEnvironment.sqlQuery("select \n" +
                "`data`['id'] id,\n" +
                "`data`['user_id'] user_id,\n" +
                "`data`['course_id'] course_id,\n" +
                "`data`['cart_price'] cart_price,\n" +
                "'1' course_num,\n" +
                "`data`['course_name'] course_name,\n" +
                "`data`['sold'] sold,\n" +
                "`data`['create_time'] create_time,\n" +
                "`data`['update_time'] update_time,\n" +
                "`ts`,\n" +
                "`pt`\n" +
                "from topic_db\n" +
                "where `table`='cart_info'\n" +
                "and ((`type`='insert') or (`type`='update' and `data`['sold']='1' and `old`['sold']='0'))");
        //tableEnvironment.executeSql(MySQLUtil.getBaseDicDDL());
        tableEnvironment.createTemporaryView("cart_add",cartAddTable);
        String targetTopic="dwd_trade_cart_add";
        tableEnvironment.executeSql("create table dwd_trade_cart_add(\n" +
                "id STRING,\n" +
                "user_id STRING,\n" +
                "course_id STRING,\n" +
                "cart_price STRING,\n" +
                "course_num STRING,\n" +
                "course_name STRING,\n" +
                "sold STRING,\n" +
                "create_time STRING,\n" +
                "update_time STRING,\n" +
                "ts BIGINT,\n" +
                "pt TIMESTAMP_LTZ(3)\n" +
                ")"+ KafkaUtil.getKafkaSinkDDL(targetTopic));
        tableEnvironment.executeSql("insert into dwd_trade_cart_add select * from cart_add");

    }
}
